package tjs.study.notes.dotest.reactor;


import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import tjs.study.notes.base.common.ThreadUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

//@SpringBootTest
public class IntegrityDemo {
    private static final Logger log = LoggerFactory.getLogger(IntegrityDemo.class);

    @Test
    public void parallel() {
        Flux.just(new UserReactor("a"), new UserReactor("b"), new UserReactor("c"))
                .parallel(2)
                .runOn(Schedulers.parallel())
                .subscribe(UserReactor::print);
    }

    @Test
    public void zip() {
        Flux.just("I", "You")
                .zipWith(Flux.just("Win", "Lose"))
                .subscribe(System.out::println);
        Flux.just("I", "You")
                .zipWith(Flux.just("Win", "Lose"),
                        (s1, s2) -> String.format("%s!%s!", s1, s2))
                .subscribe(System.out::println);
    }

    private Flux<Integer> doRpcFlux(int key) {
        return Flux.create(sink -> {
            log.info("模拟远程调用，参数为：" + key);
            ThreadUtils.sleepMilliSeconds(500 + key);
            sink.next(key);
            sink.complete();
        });
    }    
    
    @Test
    public void testParallelFlux() throws InterruptedException {
        Flux.range(1, 100)
                .parallel(6)
                .runOn(Schedulers.fromExecutor(ThreadUtils.getOne("runOn")))
//                .publishOn(Schedulers.parallel())
//                .log()
                .map((item) -> {
                    log.info("map使用的线程是谁呢：" + item);
                    return "map_" + item;
                })
                .subscribe();

        TimeUnit.MILLISECONDS.sleep(10);
    }


    @Test
    public void runOn() throws InterruptedException {
        List<Integer> targetKeys = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            targetKeys.add(i);
        }
        Flux.fromIterable(targetKeys)
                .flatMap(key -> doRpcFlux(key))
                .parallel(2)
                .runOn(Schedulers.fromExecutor(ThreadUtils.getOne("runOn")))
                .map((item) -> {
                    log.info("map使用的线程是谁呢：" + item);
                    return "map_" + item;
                }).then()
                .doOnError((item) -> System.out.println("失败了：" + item))
                .subscribe(responseData -> log.info("消费" + responseData.toString()), e -> log.info("消费error:" + e.getMessage()));
        System.out.println("任务发布结束了");
        Thread.sleep(200000);
    }
    @Test
    public void publishOn() throws InterruptedException {
        long startTime = System.currentTimeMillis();
        List<Integer> targetKeys = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            targetKeys.add(i);
        }
        Flux.fromIterable(targetKeys)
                .flatMap(key -> doRpcFlux(key))
                // publishOn下一个操作符开始使用新的线程
                // 对于一些能够在一个线程中顺序处理的任务，即使调度到ParallelScheduler上，通常也只由一个Worker来执行,改用runOn
                .publishOn(Schedulers.fromExecutor(ThreadUtils.getOne("publishOnThreadPool")))
                .map((item) -> {
                    log.info("map使用的线程是谁呢：" + item);
                    return "map_" + item;
                })
                .doOnError((item) -> System.out.println("失败了：" + item))
                .doOnComplete(() -> System.out.println("完成了"))
                .doFinally(signalType -> log.info("并发执行的时间: " + (System.currentTimeMillis() - startTime)))
                // subscribeOn只影响源头的执行环境
                .subscribeOn(Schedulers.fromExecutor(ThreadUtils.getOne("subscribeOnThreadPool")))
                .subscribe(responseData -> log.info("消费" + responseData.toString()), e -> log.info("消费error:" + e.getMessage()));
        System.out.println("任务发布结束了");
        Thread.sleep(200000);
    }
//    @Test
//    public void mergeTest7() throws InterruptedException {
//        long startTime = LocalTiker.SINGLETON.now();
//        Iterable<Integer> targetKeys = Arrays.asList( 101, 102, 103, 104, 105);
//        Flux.fromIterable(targetKeys)
//                .parallel()
//                .runOn(Schedulers.fromExecutorService(ThreadUtil.getMixedTargetThreadPool()))
//                .flatMap(key -> doRpcUseWebClient(key))
//                .sequential()
//                .doOnError(ReactorPrallelDemo::doOnError)
//                .doOnComplete(ReactorPrallelDemo::doOnComplete)
//                .doFinally(signalType -> log.info("并发执行的时间: " + LocalTiker.SINGLETON.gapFrom(startTime)))
//                .subscribe(responseData -> log.info(responseData.toString()), e -> log.info("error:" + e.getMessage()));
//        Thread.sleep(200000);
//    }
//
//    private Flux<RestOut<JSONObject>> doRpcUseWebClient(int key) {
//
//        //响应式客户端
//        WebClient client = null;
//
//        WebClient.RequestBodySpec request = null;
//
//        //方式一：极简创建
//        //方式二：使用builder创建
//        client = WebClient.builder()
//                .baseUrl("")
//                .defaultHeader(HttpHeaders.CONTENT_TYPE, "application/json")
//                .defaultHeader(HttpHeaders.USER_AGENT, "Spring 5 WebClient")
//                .build();
//        String restUrl = "http://crazy.api/demo/hello/v1?key=" + key;
//        /**
//         * 是通过 WebClient 组件构建请求
//         */
//        request = client
//                // 请求方法
//                .method(HttpMethod.GET)
//                // 请求url 和 参数
//                //                .uri(restUrl, params)
//                .uri(restUrl)
//                // 媒体的类型
//                .accept(MediaType.APPLICATION_JSON);
//
//        WebClient.ResponseSpec retrieve = request.retrieve();
//        // 处理异常 请求发出去之后判断一下返回码
//        retrieve.onStatus(status -> status.value() == 404,
//                response -> Mono.just(new RuntimeException("Not Found")));
//
//        ParameterizedTypeReference<RestOut<JSONObject>> parameterizedTypeReference =
//                new ParameterizedTypeReference<RestOut<JSONObject>>() {
//                };
//        // 返回流
//        Mono<RestOut<JSONObject>> resp = retrieve.bodyToMono(parameterizedTypeReference);
//        return Flux.from(resp);
//    }
}